package com.atguigu.day05;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import com.atguigu.utils.UrlViewCountPerWindow;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

// 使用KeyedProcessFunction实现全窗口聚合函数的功能
// 每个url在每个5秒钟的滚动窗口中被浏览的次数
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                .process(new KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>() {
                    // key: windowStartTime
                    // value: windowStartTime对应的窗口中所有元素组成的ArrayList
                    private MapState<Long, List<ClickEvent>> mapState;
                    // 窗口长度
                    private long windowSize = 5000;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mapState = getRuntimeContext().getMapState(
                                new MapStateDescriptor<Long, List<ClickEvent>>(
                                        "mapstate",
                                        Types.LONG,
                                        Types.LIST(Types.POJO(ClickEvent.class))
                                )
                        );
                    }

                    @Override
                    public void processElement(ClickEvent value, Context ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
                        // 根据数据到达的机器时间计算数据所属的窗口
                        long currTs = ctx.timerService().currentProcessingTime();
                        long windowStartTime = currTs - currTs % windowSize;
                        long windowEndTime = windowStartTime + windowSize;

                        if (!mapState.contains(windowStartTime)) {
                            // 窗口的第一条数据到达，创建一个窗口对应的新的列表
                            List<ClickEvent> events = new ArrayList<>();
                            events.add(value);
                            mapState.put(windowStartTime, events);
                        } else {
                            // 每来一条数据就把数据添加到列表中
                            mapState.get(windowStartTime).add(value);
                        }

                        // 在窗口结束时间-1毫秒注册一个定时事件
                        ctx.timerService().registerProcessingTimeTimer(windowEndTime - 1L);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
                        // 用来触发窗口的计算并销毁窗口
                        // 计算出窗口结束时间
                        long windowEndTime = timestamp + 1L;
                        long windowStartTime = windowEndTime - windowSize;
                        String url = ctx.getCurrentKey();
                        // 获取窗口中所有数据的总数
                        long count = mapState.get(windowStartTime).size();
                        // 向下游发送数据
                        out.collect(new UrlViewCountPerWindow(
                                url,
                                count,
                                windowStartTime,
                                windowEndTime
                        ));
                        // 销毁窗口
                        // .remove只清楚windowStartTime对应的键值对
                        // .clear将整个mapState清空
                        mapState.remove(windowStartTime);
                    }
                })
                .print();

        env.execute();
    }
}
